package com.doitedu.mr.day04.movie;

import com.alibaba.fastjson.JSON;
import com.doitedu.mr.day04.bean.MovieWriable;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;

/**
 * @Date 2021/12/3
 * @Created by HANGGE
 * @Description
 * 每部电影频分最高的前3部记录
 */
public class TopN02 {
    static  class TopNMapper extends Mapper<LongWritable , Text , Text, MovieWriable>{
        Text k =  new Text() ;
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            try {
                String line = value.toString();
                MovieWriable mw = JSON.parseObject(line, MovieWriable.class);
                // 获取电影的ID
                String mid = mw.getMovie();
                k.set(mid);
                context.write(k,mw);
            }catch (Exception e){
                e.printStackTrace();
            }

        }
    }

    /**
     * 如果最终的结果不是电影的均分需求, 而是电影的记录
     * 输出结果没必要是K V  所以我们的输出的K或者是V可以是 NullWritable  没有输出内容
     */
    static  class TopNReducer extends Reducer<Text, MovieWriable , MovieWriable , NullWritable>{
        /**
         * 保证相同的电影一定会聚合在一起
         * @param key
         * @param values
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void reduce(Text key, Iterable<MovieWriable> values, Context context) throws IOException, InterruptedException {
             // 创建一个存储记录的集合
            List<MovieWriable> ls = new ArrayList<MovieWriable>() ;
            // 迭代当前电影所有的评论记录
            for (MovieWriable mw : values) {// 只有一个对象
                try {
                    // 将记录存储在List集合中
                    MovieWriable movie = new MovieWriable();
                    //设置值  将数据封装在新的类中
              /*  movie.setMovie(mw.getMovie());
                movie.setRate(mw.getRate());
                movie.setTimeStamp(mw.getTimeStamp());
                movie.setUid(mw.getUid());*/
                    // 复制属性的工具类
                    BeanUtils.copyProperties(mw,movie);
                    // 添加到集合中
                    ls.add(movie) ;
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            // 对List集合排序
            Collections.sort(ls, new Comparator<MovieWriable>() {
                public int compare(MovieWriable o1, MovieWriable o2) {
                    return  Double.compare(o2.getRate(),o1.getRate());
                }
            });


            if(ls!=null && ls.size()>0){
                // 输出前3条记录
                for(int i = 0 ; i < Math.min(3,ls.size()) ; i++){
                    MovieWriable movieWriable = ls.get(i);
                    context.write(movieWriable , NullWritable.get());
                }
            }
        }
    }

    public static void main(String[] args) throws Exception {
        // Job  统计文件中单词出现的次数
        Configuration conf = new Configuration();
        // 1  创建 JOB
        Job job = Job.getInstance(conf, "topn");
        // 1) Mapper 类
        job.setMapperClass(TopNMapper.class);
        // 2) Reducer类
        job.setReducerClass(TopNReducer.class);
        // 3)  设置map输出的数据类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(MovieWriable.class);
        // 4)  设置reduce输出的数据类型
        job.setOutputKeyClass(MovieWriable.class);
        job.setOutputValueClass(NullWritable.class);
        // 5) 设置输入的数据路径
        FileInputFormat.setInputPaths(job , new Path("E:\\mrdata\\movie\\input"));
        // 6) 设置输出的结果保存路径
        FileOutputFormat.setOutputPath(job, new Path("E:\\mrdata\\movie\\topn03"));
        // 2 提交工作
        // 等待执行完成 , 打印执行过
        job.waitForCompletion(true) ;
    }

}
